package com.sparkStreaming.project.spark


import com.sparkStreaming.project.dao.{CourseClickCountDAO, CourseSearchClickCountDAO}
import com.sparkStreaming.project.entity.{ClickLog, CourseClickCount, CourseSearchClickCount}
import com.sparkStreaming.project.utils.DateUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

/**
  * 使用Spark Streaming处理Kafka过来的数据
  */
object StatStreamingApp_1 {

  def main(args: Array[String]): Unit = {

    if (args.length != 4) {
      println("Usage: 2019StatStreamingApp <zkQuorum> <group> <topics> <numThreads>")
      System.exit(1)
    }

    //hadoop000:2181 test streamingtopic 1
    val Array(zkQuorum, groupId, topics, numThreads) = args

    val sparkConf = new SparkConf().setAppName("2019StatStreamingApp") //.setMaster("local[5]")
    val ssc = new StreamingContext(sparkConf, Seconds(60))

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

    val messages = KafkaUtils.createStream(ssc, zkQuorum, groupId, topicMap)

    // 测试步骤一：测试数据接收
    //messages.map(_._2).count().print

    // 测试步骤二：数据清洗
    val logs = messages.map(_._2)
    //line   ==  55.187.10.168	2019-04-16 09:35:01	"GET /class/131.html HTTP/1.1"	200	https://www.sogou.com/web?query=大数据面试
    val cleanData = logs.map(line => {
      val infos = line.split("\t")

      // infos(2) = "GET /class/130.html HTTP/1.1"
      // url = /class/130.html
      val url = infos(2).split(" ")(1)
      var courseId = 0

      // 把实战课程的课程编号拿到了
      if (url.startsWith("/class")) {
        val courseIdHTML = url.split("/")(2)
        courseId = courseIdHTML.substring(0, courseIdHTML.lastIndexOf(".")).toInt
      }

      ClickLog(infos(0), DateUtils.parseToMinute(infos(1)), courseId, infos(3).toInt, infos(4))
    }).filter(clicklog => clicklog.courseId != 0)


        cleanData.print()

    ssc.start()
    ssc.awaitTermination()
  }

}
